package zip;

import org.apache.commons.net.ftp.FTPClient;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;

import java.io.IOException;
import java.io.OutputStream;
import java.net.URI;

/**
 * Created by ye on 2016/3/17.
 */
public class HdfsReader {
    private String path;
    private Configuration conf;
    private FileSystem fs;
    FtpConn ftpConn = new FtpConn();
    public HdfsReader() throws IOException {
        conf = new Configuration();
        fs = FileSystem.get(conf);
    }
    public HdfsReader(String path) throws IOException {
        this.path = path;
        conf = new Configuration();
        fs = FileSystem.get(URI.create(path), conf);
    }

    public boolean run(String output) throws Exception {

        FileStatus fileList[] = fs.listStatus(new Path(path));
        FSDataInputStream hdfsInStream;
        OutputStream out;
        OutputStream gzipOut;
        FTPClient ftp = ftpConn.connectFtp("ftp.properties");
        for(FileStatus fileStatus : fileList){
            hdfsInStream = fs.open(fileStatus.getPath());
            System.out.println(fileStatus.getPath().toString());
//            out = new FileOutputStream(output+fileStatus.getPath().getName()+".gz");
            gzipOut = ftp.storeFileStream(fileStatus.getPath().getName()+".gz");
            byte[] ioBuffer = new byte[1024];
            int readLen = hdfsInStream.read(ioBuffer);
            String header = "this is a header.\n";
            gzipOut.write(header.getBytes());
            while(-1 != readLen){
                gzipOut.write(ioBuffer, 0, readLen);
                readLen = hdfsInStream.read(ioBuffer);
            }
            String tail = "\nthis is a tail.\n";
            gzipOut.write(tail.getBytes());
            gzipOut.close();
            hdfsInStream.close();

        }
        fs.close();
        ftpConn.closeFtp();
        System.out.println("done");
        return true;
    }

    public String getPath() {
        return path;
    }

    public void setPath(String path) {
        this.path = path;
    }
}
